package cn.doitedu.hbase.coprosessor;

import com.google.protobuf.Service;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.coprocessor.ObserverContext;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessor;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
import org.apache.hadoop.hbase.coprocessor.RegionObserver;
import org.apache.hadoop.hbase.wal.WALEdit;

import java.io.IOException;
import java.util.List;
import java.util.Map;
import java.util.NavigableMap;
import java.util.Optional;

/**
 * @date: 2019/7/12
 * @site: www.doitedu.cn
 * @author: hunter.d 涛哥
 * @qq: 657270652
 * @description: 实验准备：
 * 0. 代码打成jar包，取名cop.jar ，上传到 hdfs://[你的namenode]:[你的namenode端口]/copjar/cop.jar
 * 1. hbase建表
 * create 't_star','f'
 * create 't_fans','f'
 * 2. disable 't_star'
 * 3. hbase> alter 't_star','coprocessor'=>'hdfs://spark01:8020/copjar/cop.jar|cn.doitedu.hbase.coprosessor.FansCoprocessor|1001|'
 * 4. enable 't_star'
 * 5. put 't_star','A','f:B','B'
 * 6. 检查结果
 */
public class FansCoprocessor implements RegionCoprocessor, RegionObserver {
    Connection conn = null;
    Table t_fans =null;

    @Override
    public Optional<RegionObserver> getRegionObserver() {

        return Optional.of(this);
    }

    /**
     * 拦截的具体逻辑,拦截的是put事件，工作在put动作执行之前
     *
     * @param c
     * @param put
     * @param edit
     * @param durability
     * @throws IOException
     */
    @Override
    public void prePut(ObserverContext<RegionCoprocessorEnvironment> c, Put put, WALEdit edit, Durability durability) throws IOException {

        // 1. 拆解客户端的put中数据，得到  uid 和 关注目标的uid

        // 2.创建一个到粉丝表的连接

        // 3.构造一个新的put对象（被关注者，粉丝）

        // 4.用粉丝表连接插入数据

        // 5.放开

        Log log = LogFactory.getLog(FansCoprocessor.class);
        log.error("-------------------entered coprocessor-------------------------------------");
        // 解剖拦截到的put数据内容  rowkey：用户id    列族：f    column： 被关注者   value：被关注者[或者是null]
        // 获取行键，就是用户id
        // 获取列名，就是被关注者（偶像）
        NavigableMap<byte[], List<Cell>> fm = put.getFamilyCellMap();
        Map.Entry<byte[], List<Cell>> e = fm.firstEntry();


        List<Cell> cells = e.getValue();
        Cell ce = cells.get(0);
        byte[] uid = CellUtil.cloneRow(ce);
        byte[] s = CellUtil.cloneQualifier(ce);

        log.error("-------------------uid-------------------------------------" + new String(uid));
        log.error("-------------------star-------------------------------------" + new String(s));

        // 反转生成一个反向关注数据：   rowkey：被关注者   列族：f   column:用户id   value：用户id[或者null]

        Put put1 = new Put(s);
        put1.addColumn("f".getBytes(), uid, uid);

        t_fans.put(put1);
        log.error("-------------------put is over -------------------------------------");



    }


    @Override
    public void start(CoprocessorEnvironment env) throws IOException {
        // 然后将反向数据插入“被关注”索引表
        Configuration conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", "spark01:2181,spark02:2181,spark03:2181");

        conn = ConnectionFactory.createConnection(conf);
        t_fans = conn.getTable(TableName.valueOf("t_fans"));
    }

    @Override
    public void stop(CoprocessorEnvironment env) throws IOException {
        t_fans.close();
        conn.close();

    }
}
